package cn.ucox.web.ms.polling;

import cn.ucox.web.ms.config.DataExchangeConfig.PollingConfig;
import cn.ucox.web.ms.packer.TransactionDataProcessor;
import cn.ucox.web.ms.redis.RedisClient;
import org.springframework.data.redis.core.RedisTemplate;

import java.util.HashMap;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import static cn.ucox.web.ms.Constant.*;

/**
 * Bucket数据处理
 *
 * @author chenw
 * @create 2017-05-11 15:59
 * @email javacspring@gmail.com
 */
public class BucketDataProcessor {

    private PollingConfig mPollingConfig;

    private static RedisClient redisClient;

    static {
        redisClient = RedisClient.instance();
    }

    private ExecutorService service = Executors.newFixedThreadPool(CPU_PROCESSOR_NUMBER * 2);

    private RedisTemplate<String, Object> mRedisTemplate;
    private IFileLoader mFileLoader;

    public BucketDataProcessor(PollingConfig pollingConfig,
                               IFileLoader fileLoader) {
        this.mPollingConfig = pollingConfig;
        this.mFileLoader = fileLoader;
    }

    public void postBucketData(SyncAnchor syncAnchor, List<SyncData> syncData) {
        //根据当前初始化任务处理器个数，分割列表数据，并投递至任务处理器
        int dataRows = syncData.size();
        //根据任务量拆分子任务号数量
        int offset = (int) mPollingConfig.getProcessNumPerCpuThread();
        int taskNumber = (0 == dataRows % offset) ? (dataRows / offset) : (dataRows / offset + 1);

        //刷新任务报告
        syncAnchor.setTotalTaskNum(taskNumber);
        syncAnchor.setTransactionDataRows(dataRows);

        redisClient.hmset(SYNC_ANCHOR, new HashMap<String, String>() {{
            put(SYNC_ANCHOR_TOTAL_TASK_NUM, String.valueOf(taskNumber));
            put(SYNC_ANCHOR_TRANSACTION_DATA_ROWS, String.valueOf(dataRows));
            put(SYNC_ANCHOR_TRANSACTION_DONE,"N");
        }});

        for (int i = 0; i < taskNumber; i++) {
            //将附件数据与与文本数据分为单独的处理器进行处理
            int start = i * offset;
            int end = (start + offset);
            end = (end > syncData.size() ? syncData.size() : end);
            service.submit(new TransactionDataProcessor(
                    syncAnchor,
                    mFileLoader,
                    syncData.subList(start, end)));
        }
    }
}
